// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <cstdint>
#include <memory>
#include <string>
#include <string_view>
#include <vector>

#include <nebula/io/type_fwd.h>
#include <nebula/future/cancel.h>
#include <nebula/util/type_fwd.h>
#include <turbo/base/macros.h>
#include <nebula/future/type_fwd.h>
#include <nebula/types/type_fwd.h>

namespace nebula::io {

    struct ReadRange {
        int64_t offset;
        int64_t length;

        friend bool operator==(const ReadRange &left, const ReadRange &right) {
            return (left.offset == right.offset && left.length == right.length);
        }

        friend bool operator!=(const ReadRange &left, const ReadRange &right) {
            return !(left == right);
        }

        bool Contains(const ReadRange &other) const {
            return (offset <= other.offset && offset + length >= other.offset + other.length);
        }
    };

    /// EXPERIMENTAL: options provider for IO tasks
    ///
    /// Includes an Executor (which will be used to execute asynchronous reads),
    /// a MemoryPool (which will be used to allocate buffers when zero copy reads
    /// are not possible), and an external id (in case the executor receives tasks from
    /// multiple sources and must distinguish tasks associated with this IOContext).
    struct TURBO_EXPORT IOContext {
        // No specified executor: will use a global IO thread pool
        IOContext() : IOContext(default_memory_pool(), StopToken::unstoppable()) {}

        explicit IOContext(StopToken stop_token)
                : IOContext(default_memory_pool(), std::move(stop_token)) {}

        explicit IOContext(MemoryPool *pool, StopToken stop_token = StopToken::unstoppable());

        explicit IOContext(MemoryPool *pool, ::nebula::internal::Executor *executor,
                           StopToken stop_token = StopToken::unstoppable(),
                           int64_t external_id = -1)
                : pool_(pool),
                  executor_(executor),
                  external_id_(external_id),
                  stop_token_(std::move(stop_token)) {}

        explicit IOContext(::nebula::internal::Executor *executor,
                           StopToken stop_token = StopToken::unstoppable(),
                           int64_t external_id = -1)
                : pool_(default_memory_pool()),
                  executor_(executor),
                  external_id_(external_id),
                  stop_token_(std::move(stop_token)) {}

        MemoryPool *pool() const { return pool_; }

        ::nebula::internal::Executor *executor() const { return executor_; }

        // An application-specific ID, forwarded to executor task submissions
        int64_t external_id() const { return external_id_; }

        StopToken stop_token() const { return stop_token_; }

    private:
        MemoryPool *pool_;
        ::nebula::internal::Executor *executor_;
        int64_t external_id_;
        StopToken stop_token_;
    };

    class TURBO_EXPORT FileInterface : public std::enable_shared_from_this<FileInterface> {
    public:
        virtual ~FileInterface() = 0;

        /// \brief close the stream cleanly
        ///
        /// For writable streams, this will attempt to flush any pending data
        /// before releasing the underlying resource.
        ///
        /// After close() is called, closed() returns true and the stream is not
        /// available for further operations.
        virtual turbo::Status close() = 0;

        /// \brief close the stream asynchronously
        ///
        /// By default, this will just submit the synchronous close() to the
        /// default I/O thread pool. Subclasses may implement this in a more
        /// efficient manner.
        virtual Future<> close_async();

        /// \brief close the stream abruptly
        ///
        /// This method does not guarantee that any pending data is flushed.
        /// It merely releases any underlying resource used by the stream for
        /// its operation.
        ///
        /// After abort() is called, closed() returns true and the stream is not
        /// available for further operations.
        virtual turbo::Status abort();

        /// \brief Return the position in this stream
        virtual turbo::Result<int64_t> tell() const = 0;

        /// \brief Return whether the stream is closed
        virtual bool closed() const = 0;

        FileMode::type mode() const { return mode_; }

    protected:
        FileInterface() : mode_(FileMode::READ) {}

        FileMode::type mode_;

        void set_mode(FileMode::type mode) { mode_ = mode; }

    private:
        TURBO_DISALLOW_COPY_AND_ASSIGN(FileInterface);
    };

    class TURBO_EXPORT Seekable {
    public:
        virtual ~Seekable() = default;

        virtual turbo::Status Seek(int64_t position) = 0;
    };

    class TURBO_EXPORT Writable {
    public:
        virtual ~Writable() = default;

        /// \brief write the given data to the stream
        ///
        /// This method always processes the bytes in full.  Depending on the
        /// semantics of the stream, the data may be written out immediately,
        /// held in a buffer, or written asynchronously.  In the case where
        /// the stream buffers the data, it will be copied.  To avoid potentially
        /// large copies, use the write variant that takes an owned Buffer.
        virtual turbo::Status write(const void *data, int64_t nbytes) = 0;

        /// \brief write the given data to the stream
        ///
        /// Since the Buffer owns its memory, this method can avoid a copy if
        /// buffering is required.  See write(const void*, int64_t) for details.
        virtual turbo::Status write(const std::shared_ptr<Buffer> &data);

        virtual turbo::Status write(BufferSpan data);

        /// \brief flush buffered bytes, if any
        virtual turbo::Status flush();

        turbo::Status write(std::string_view data);
    };

    class TURBO_EXPORT Readable {
    public:
        virtual ~Readable() = default;

        /// \brief read data from current file position.
        ///
        /// read at most `nbytes` from the current file position into `out`.
        /// The number of bytes read is returned.
        virtual turbo::Result<int64_t> read(int64_t nbytes, void *out) = 0;

        /// \brief read data from current file position.
        ///
        /// read at most `nbytes` from the current file position. Less bytes may
        /// be read if EOF is reached. This method updates the current file position.
        ///
        /// In some cases (e.g. a memory-mapped file), this method may avoid a
        /// memory copy.
        virtual turbo::Result<std::shared_ptr<Buffer>> read(int64_t nbytes) = 0;

        /// EXPERIMENTAL: The IOContext associated with this file.
        ///
        /// By default, this is the same as default_io_context(), but it may be
        /// overridden by subclasses.
        virtual const IOContext &io_context() const;
    };

    class TURBO_EXPORT OutputStream : virtual public FileInterface, public Writable {
    protected:
        OutputStream() = default;
    };

    class TURBO_EXPORT InputStream : virtual public FileInterface, virtual public Readable {
    public:
        /// \brief advance or skip stream indicated number of bytes
        /// \param[in] nbytes the number to move forward
        /// \return turbo::Status
        turbo::Status advance(int64_t nbytes);

        /// \brief Return zero-copy string_view to upcoming bytes.
        ///
        /// Do not modify the stream position.  The view becomes invalid after
        /// any operation on the stream.  May trigger buffering if the requested
        /// size is larger than the number of buffered bytes.
        ///
        /// May return NotImplemented on streams that don't support it.
        ///
        /// \param[in] nbytes the maximum number of bytes to see
        virtual turbo::Result<std::string_view> peek(int64_t nbytes);

        /// \brief Return true if InputStream is capable of zero copy Buffer reads
        ///
        /// Zero copy reads imply the use of Buffer-returning read() overloads.
        virtual bool supports_zero_copy() const;

        /// \brief read and return stream metadata
        ///
        /// If the stream implementation doesn't support metadata, empty metadata
        /// is returned.  Note that it is allowed to return a null pointer rather
        /// than an allocated empty metadata.
        virtual turbo::Result<std::shared_ptr<const KeyValueMetadata>> read_metadata();

        /// \brief read stream metadata asynchronously
        virtual Future<std::shared_ptr<const KeyValueMetadata>> read_metadata_async(
                const IOContext &io_context);

        Future<std::shared_ptr<const KeyValueMetadata>> read_metadata_async();

    protected:
        InputStream() = default;
    };

    class TURBO_EXPORT RandomAccessFile : public InputStream, public Seekable {
    public:
        /// Necessary because we hold a std::unique_ptr
        ~RandomAccessFile() override;

        /// \brief create an isolated InputStream that reads a segment of a
        /// RandomAccessFile. Multiple such stream can be created and used
        /// independently without interference
        /// \param[in] file a file instance
        /// \param[in] file_offset the starting position in the file
        /// \param[in] nbytes the extent of bytes to read. The file should have
        /// sufficient bytes available
        static turbo::Result<std::shared_ptr<InputStream>> get_stream(
                std::shared_ptr<RandomAccessFile> file, int64_t file_offset, int64_t nbytes);

        /// \brief Return the total file size in bytes.
        ///
        /// This method does not read or move the current file position, so is safe
        /// to call concurrently with e.g. read_at().
        virtual turbo::Result<int64_t> get_size() = 0;

        /// \brief read data from given file position.
        ///
        /// At most `nbytes` bytes are read.  The number of bytes read is returned
        /// (it can be less than `nbytes` if EOF is reached).
        ///
        /// This method can be safely called from multiple threads concurrently.
        /// It is unspecified whether this method updates the file position or not.
        ///
        /// The default RandomAccessFile-provided implementation uses Seek() and read(),
        /// but subclasses may override it with a more efficient implementation
        /// that doesn't depend on implicit file positioning.
        ///
        /// \param[in] position Where to read bytes from
        /// \param[in] nbytes The number of bytes to read
        /// \param[out] out The buffer to read bytes into
        /// \return The number of bytes read, or an error
        virtual turbo::Result<int64_t> read_at(int64_t position, int64_t nbytes, void *out);

        /// \brief read data from given file position.
        ///
        /// At most `nbytes` bytes are read, but it can be less if EOF is reached.
        ///
        /// \param[in] position Where to read bytes from
        /// \param[in] nbytes The number of bytes to read
        /// \return A buffer containing the bytes read, or an error
        virtual turbo::Result<std::shared_ptr<Buffer>> read_at(int64_t position, int64_t nbytes);

        /// EXPERIMENTAL: read data asynchronously.
        virtual Future<std::shared_ptr<Buffer>> read_async(const IOContext &, int64_t position,
                                                          int64_t nbytes);

        /// EXPERIMENTAL: read data asynchronously, using the file's IOContext.
        Future<std::shared_ptr<Buffer>> read_async(int64_t position, int64_t nbytes);

        /// EXPERIMENTAL: Explicit multi-read.
        /// \brief Request multiple reads at once
        ///
        /// The underlying filesystem may optimize these reads by coalescing small reads into
        /// large reads or by breaking up large reads into multiple parallel smaller reads.  The
        /// reads should be issued in parallel if it makes sense for the filesystem.
        ///
        /// One future will be returned for each input read range.  Multiple returned futures
        /// may correspond to a single read.  Or, a single returned future may be a combined
        /// result of several individual reads.
        ///
        /// \param[in] ranges The ranges to read
        /// \return A future that will complete with the data from the requested range is
        /// available
        virtual std::vector<Future<std::shared_ptr<Buffer>>> read_many_async(
                const IOContext &, const std::vector<ReadRange> &ranges);

        /// EXPERIMENTAL: Explicit multi-read, using the file's IOContext.
        std::vector<Future<std::shared_ptr<Buffer>>> read_many_async(
                const std::vector<ReadRange> &ranges);

        /// EXPERIMENTAL: Inform that the given ranges may be read soon.
        ///
        /// Some implementations might arrange to prefetch some of the data.
        /// However, no guarantee is made and the default implementation does nothing.
        /// For robust prefetching, use read_at() or read_async().
        virtual turbo::Status will_need(const std::vector<ReadRange> &ranges);

    protected:
        RandomAccessFile();

    private:
        struct TURBO_NO_EXPORT Impl;
        std::unique_ptr<Impl> interface_impl_;
    };

    class TURBO_EXPORT WritableFile : public OutputStream, public Seekable {
    public:
        virtual turbo::Status write_at(int64_t position, const void *data, int64_t nbytes) = 0;

    protected:
        WritableFile() = default;
    };

    class TURBO_EXPORT ReadWriteFileInterface : public RandomAccessFile, public WritableFile {
    protected:
        ReadWriteFileInterface() { RandomAccessFile::set_mode(FileMode::READWRITE); }
    };

    /// \brief Return an iterator on an input stream
    ///
    /// The iterator yields a fixed-size block on each next() call, except the
    /// last block in the stream which may be smaller.
    /// Once the end of stream is reached, next() returns nullptr
    /// (unlike InputStream::read() which returns an empty buffer).
    TURBO_EXPORT
    turbo::Result<turbo::Iterator<std::shared_ptr<Buffer>>> make_input_stream_iterator(
            std::shared_ptr<InputStream> stream, int64_t block_size);

}  // namespace nebula::io

